package com.czk.springboot.reactive.reactor.flux;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

import java.time.Duration;

/**
 * 自定义消费者
 */

public class FluxDemo3 {
    public static void main(String[] args) {

        Flux<Integer> flux = Flux.range(0, 10)
                .delayElements(Duration.ofSeconds(1))
                        .doOnCancel(()-> System.out.println("流被取消了订阅了===="));


       // flux.subscribe(e-> System.out.println(e));
        flux.subscribe(new BaseSubscriber<>() {
            @Override
            protected Subscription upstream() {
                return super.upstream();
            }

            @Override
            public boolean isDisposed() {
                return super.isDisposed();
            }

            @Override
            public void dispose() {
                super.dispose();
            }

            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                System.out.println("==订阅者与发布者绑定好了====");
                // 像发布者索要一个元素
                super.request(1); //背压模式

            }

            @Override
            protected void hookOnNext(Integer value) {
                System.out.println("元素到达了===" + value);
                // 再索要元素
                if (value < 5) {
                    super.request(1);
                } else {
                    // 取消订阅
                    super.cancel();
                }
            }

            @Override
            protected void hookOnComplete() {
                super.hookOnComplete();
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                super.hookOnError(throwable);
            }

            @Override
            protected void hookOnCancel() {
                super.hookOnCancel();
            }

            @Override
            protected void hookFinally(SignalType type) {
                super.hookFinally(type);
            }
        });
        while (true){}
    }
}
